package org.pocs.akka.untyped;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.pocs.common.dto.Signal;

import akka.actor.Props;
import akka.actor.UntypedActor;
import akka.japi.Creator;

public class WorkerActor extends UntypedActor {

	private Signal signal;
	
	public static Props getProps(Signal signal) {
		return Props.create(new WorkerActor.ActorCreator(signal));
	}
	
	public WorkerActor(Signal signal) {
		super();
		this.setSignal(signal);
	}

	@Override
	public void onReceive(Object arg0) throws Exception {
		
		// Processing. From here we can start our project logic
		
		
		
		
		ActorCreator.objectPool.put(signal.getSourceNode(), this);
		getSender().tell(CurrentStatus.ACTOR_SUCCESS, getSelf());
	}	

	public Signal getSignal() {
		return signal;
	}

	public void setSignal(Signal signal) {
		this.signal = signal;
	}

	static class ActorCreator implements Creator<WorkerActor> {
	
		private static final long serialVersionUID = 1L;
		private static Map<String, WorkerActor> objectPool = new HashMap<>();
		
		private Signal signal;
		
		public ActorCreator(Signal signal) {
			super();
			this.signal = signal;
		}
	
		@Override
		public WorkerActor create() throws Exception {
			// Fetch Worker actor from pool
	       
			if(objectPool.containsKey(signal.getSourceNode())) {
	        
				WorkerActor workerActor = objectPool.get(signal.getSourceNode());
	        	workerActor.setSignal(signal);
	        	
	        	return workerActor;
	        
			} else {
	        	return new WorkerActor(signal);
	        }
		}
	}		
}